package org.example.flink.source;

import cn.hutool.core.date.DateUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.pulsar.source.PulsarSource;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.pulsar.client.api.SubscriptionType;

/**
 * @author zhangjungang
 * @desc test
 * @date 2021/12/15 下午4:16
 */
public class Flink14 {


    public static void main(String[] args) {

        String serviceUrl = "http://127.0.0.1:8080";
        String adminUrl = serviceUrl;

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        PulsarSource<String> pulsarSource = PulsarSource.builder()
                .setServiceUrl(serviceUrl)
                .setAdminUrl(adminUrl)
                .setStartCursor(StartCursor.earliest())
                .setTopics("my-topic")
                .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
                .setSubscriptionName("my-subscription")
                .setSubscriptionType(SubscriptionType.Shared)
                .build();

        DataStreamSource<String> pulsar_source =
                env.fromSource(pulsarSource, WatermarkStrategy.noWatermarks(), "Pulsar Source");

        pulsar_source.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                System.out.println("-----"+value+" now="+ DateUtil.now());
            }
        });

        try {
            env.execute("xxxxx");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
